# -*- coding: utf-8 -*-
from datetime import timedelta
from utils.operators.spark_submit import SparkSubmitOperator
from utils.operators.cluster_for_spark_sql_operator import SparkSqlOperator
from jms.time_sensor import time_after_07_10

jdbcUrl='{{ var.json.oracle_bigdata03.url }}'
username='{{ var.json.oracle_bigdata03.username }}'
password='{{ var.json.oracle_bigdata03.password }}'
nowdt='{{ execution_date | cst_ds }}'
nextdt='{{ execution_date | date_add(1) | cst_ds }}'
table="yl_oms_interceptorpiece"
env='{{ var.value.env_sync }}'

jsonpara="""{
"reader":{
"connect":{
"url":"jdbcUrlpara",
"username":"usernamepara",
"password":"passwordpara",
"driver":"oracle.jdbc.driver.OracleDriver"
},
"dbtype":"oracle",
"tableName":"tablepara",
"where":"update_time >= to_date('nextdt 00:00:00', 'yyyy-mm-dd hh24:mi:ss') and update_time < to_date('nextdt 07:30:00', 'yyyy-mm-dd hh24:mi:ss')",
"query":"",
"splitColumn":"id",
"equalitySectioning":1,
"containsnull":0,
"fetchsize":"1024",
"threadNumber":3
},
"channel":{
"filterAbnormalCharacter":1
},
"writer":{
"dbtype":"hive",
"tableName":"yl_oms_interceptorpiece_di",
"database":"jms_ods",
"writeMode": "overwrite",
"partitionColumn":"dt",
"partitionValue":"nextdt"},
"settting":{
"env":"envpara"}
}""".replace("jdbcUrlpara",jdbcUrl).replace("usernamepara",username).replace("passwordpara",password). \
    replace("nowdt",nowdt).replace("nextdt",nextdt). \
    replace("tablepara",table). \
    replace("envpara",env)

# from airflow.operators.dummy_operator import DummyOperator
# jms_ods__yl_oms_interceptorpiece__0730 = DummyOperator(
#     task_id='jms_ods__yl_oms_interceptorpiece__0730',
#     email='chenhongping@yl-scm.com',
#     retries=0,
#     priority_weight=0,
#     # sla=timedelta(hours=2),
# )
jms_ods__yl_oms_interceptorpiece_di__0730 = SparkSubmitOperator(
    task_id='jms_ods__yl_oms_interceptorpiece_di__0730',
    email=['rongguangfan@jtexpress.com','yl_bigdata@yl-scm.com'],
    name='jms_ods__yl_oms_interceptorpiece_di__0730_{{ execution_date | date_add(1) | cst_ds }}',
    pool='oracle_tab',
    pool_slots=3,
    execution_timeout=timedelta(hours=1),
    driver_memory='2G',
    executor_memory='2G',
    executor_cores=2,
    num_executors=1,
    conf={'spark.dynamicAllocation.enabled'                  : 'true',
          'spark.shuffle.service.enabled'                    : 'true',
          'spark.dynamicAllocation.maxExecutors'             : 3,
          'spark.dynamicAllocation.cachedExecutorIdleTimeout': 30,
          'spark.sql.sources.partitionOverwriteMode'         : 'dynamic',
          'spark.executor.memoryOverhead'                    : '1G',
          },
    java_class='com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver',  # spark 主类
    application='hdfs:///scheduler/jms/spark/sync/spark_sync.jar',  # spark jar 包
    application_args=[jsonpara,],
)

jms_ods__yl_oms_interceptorpiece__0730 = SparkSqlOperator(
    task_id='jms_ods__yl_oms_interceptorpiece__0730',
    email=['rongguangfan@jtexpress.com','yl_bigdata@yl-scm.com'],
    name='jms_ods__yl_oms_interceptorpiece__0730_{{ execution_date | date_add(1) | cst_ds }}',
    sql='jms/ods/oms/yl_oms_interceptorpiece__0730/execute.sql',
    pool_slots=2,
    executor_cores=2,
    executor_memory='6G',
    driver_memory='2G',
    num_executors=10,  # spark.dynamicAllocation.enabled 为 True 时，num_executors 表示最少 Executor 数
    conf={'spark.dynamicAllocation.maxExecutors': 40,
          'spark.sql.sources.partitionOverwriteMode': 'dynamic',
          'spark.dynamicallocation.enabled': 'true',
          'spark.dynamicAllocation.cachedExecutorIdleTimeout': 120,
          'spark.shuffle.service.enabled': 'true',
          'spark.sql.shuffle.partitions': 200
          },
    hiveconf={'hive.exec.dynamic.partition': 'true',
              'hive.exec.dynamic.partition.mode': 'nonstrict',
              'hive.exec.max.dynamic.partitions.pernode': 20,
              'hive.exec.max.dynamic.partitions': 20
              },
    yarn_queue='pro',
    execution_timeout=timedelta(minutes=40),
)

jms_ods__yl_oms_interceptorpiece__0730 << jms_ods__yl_oms_interceptorpiece_di__0730 << time_after_07_10
